package cn.zhaoyuening.fanrpc.consumer.core;

import cn.zhaoyuening.fanrpc.common.beans.RpcRequest;
import cn.zhaoyuening.fanrpc.common.beans.RpcResponce;
import cn.zhaoyuening.fanrpc.common.beans.RpcServiceAddress;
import cn.zhaoyuening.fanrpc.common.utils.MarshallingCodeCFactory;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.net.InetSocketAddress;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

public class RpcRequestSender {
    private Map<RpcServiceAddress, Channel> rpcChannelMap = new HashMap<RpcServiceAddress, Channel>();
    private Bootstrap bootstrap;
    private EventLoopGroup group;
//    private Executor executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
    public void start() {
        group = new NioEventLoopGroup();
        bootstrap = new Bootstrap();
        bootstrap.group(group)
                .channel(NioSocketChannel.class)
                .option(ChannelOption.SO_KEEPALIVE,true)
                .handler(new ChannelInitializer<Channel>() {
                    @Override
                    protected void initChannel(Channel ch) throws Exception {
                        ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingDecoder());
                        ch.pipeline().addLast(MarshallingCodeCFactory.buildMarshallingEncoder());
                    }
                });
    }

    public RpcResponce sendRequest(final RpcServiceAddress address, final RpcRequest request) {
        Channel channel = rpcChannelMap.get(address);
        final CountDownLatch cdl = new CountDownLatch(1);
        if (channel != null && !channel.isOpen()) {
            channel = null;
        }
        if (channel == null) {
            channel = bootstrap.connect(address.getAddress(), address.getPort()).channel();
            rpcChannelMap.put(address, channel);
        }

        final Channel c = channel;
        final RpcResponce[] resp = new RpcResponce[1];
        bootstrap.handler(new ChannelHandlerAdapter(){
            @Override
            public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
                request.setReqId(UUID.randomUUID().toString());
                c.write(request);
            }


            @Override
            public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                if (msg instanceof RpcResponce) {
                    RpcResponce responce = (RpcResponce) msg;
                    if (responce.getRespId().equals(request.getReqId())) {
                        resp[0] = responce;
                        cdl.countDown();
                    }
                }
            }

        });
        try {
            cdl.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return resp[0];
    }


    public void end() {
        group.shutdownGracefully();
    }
}
